package com.heima.article.stream;

import com.alibaba.fastjson.JSON;
import com.heima.common.constants.HotArticleConstants;
import com.heima.model.mess.ArticleVisitStreamMess;
import com.heima.model.mess.UpdateArticleMess;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.time.Duration;

@Configuration
@Slf4j
public class HotArticleStreamHandler {
    @Bean
    public KStream<String, String> kStream(StreamsBuilder streamsBuilder){
        //接收消息
        KStream<String, String> stream = streamsBuilder.stream(HotArticleConstants.HOT_ARTICLE_SCORE_TOPIC);
        //聚合流式处理
        stream.map((key, value) -> {
            UpdateArticleMess mess = JSON.parseObject(value, UpdateArticleMess.class);    //将 json 形式的消息转成 UpdateArticleMess
            //重置消息的              key(文章的id):1234343434                 value: likes:1
            return new KeyValue<>(mess.getArticleId().toString(), mess.getType().name() + " : " + mess.getAdd());
        })
                .groupBy((key, value) -> key)    //按照文章id进行聚合
                .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))    //时间窗口, 10s一次
                /**
                 * 自行的完成聚合的计算
                 */
                .aggregate(new Initializer<String>() {
                    /**
                     * 初始方法，返回值是消息的value
                     */
                    @Override
                    public String apply() {
                        return "COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0";    //初始值：收藏、评论、点赞、浏览都设置为0
                    }
                    /**
                     * 真正的聚合操作，返回值是消息的value
                     */
                }, new Aggregator<String, String, String>() {
                    @Override
                    public String apply(String key, String value, String aggValue){ //key，value是上面重置消息的key value，aggValue是apply方法的初始值
                        if(StringUtils.isBlank(value)){
                            return aggValue;   //如果value为空，直接返回初始值 aggValue("COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0")
                        }
                        String[] aggAry = aggValue.split(",");    //对初始值进行以","分割   结果如：COLLECTION:0
                        int col=0, com=0, lik=0, vie=0;
                        for (String agg : aggAry) {
                            String[] spl = agg.split(":");    //再将结果进行以 ":" 分割   结果如：COLLECTION 和 0
                            /**
                             * 获得初始值，也是时间窗口内计算之后的值
                             */
                            switch (UpdateArticleMess.UpdateArticleType.valueOf(spl[0])){    //拿到spl中的第[0]参数，结果如：COLLECTION
                                case COLLECTION:    //如果第[0]参数是收藏
                                    col = Integer.parseInt(spl[1]);    //就把spl中的第[1]个参数赋值给收藏，这个参数就如 COLLECTION:0 中的 0
                                    break;
                                case COMMENT:    //如果第[0]参数是评论
                                    com = Integer.parseInt(spl[1]);    //就把spl中的第[1]个参数赋值给评论，这个参数就如 COLLECTION:0 中的 0
                                    break;
                                case LIKES:    //如果第[0]参数是点赞
                                    lik = Integer.parseInt(spl[1]);    //就把spl中的第[1]个参数赋值给点赞，这个参数就如 COLLECTION:0 中的 0
                                    break;
                                case VIEWS:    //如果第[0]参数是浏览
                                    vie = Integer.parseInt(spl[1]);    //就把spl中的第[1]个参数赋值给浏览，这个参数就如 COLLECTION:0 中的 0
                                    break;
                            }}
                        /**
                         * 累加操作
                         */
                        String[] valAry = value.split(":");    //对value进行以 ":" 分割，结果如：COLLECTION 和 0
                        switch (UpdateArticleMess.UpdateArticleType.valueOf(valAry[0])){    //拿到valAry中的第[0]参数，结果如：COLLECTION
                            case COLLECTION:    //如果第[0]参数是收藏
                                col += Integer.parseInt(valAry[1]);    //就把spl中的第[1]个参数累加给收藏，这个参数就如 COLLECTION:0 中的 0
                                break;
                            case COMMENT:    //如果第[0]参数是评论
                                com += Integer.parseInt(valAry[1]);    //就把spl中的第[1]个参数累加给评论，这个参数就如 COLLECTION:0 中的 0
                                break;
                            case LIKES:    //如果第[0]参数是点赞
                                lik += Integer.parseInt(valAry[1]);    //就把spl中的第[1]个参数累加给点赞，这个参数就如 COLLECTION:0 中的 0
                                break;
                            case VIEWS:    //如果第[0]参数是浏览
                                vie += Integer.parseInt(valAry[1]);    //就把spl中的第[1]个参数累加给浏览，这个参数就如 COLLECTION:0 中的 0
                                break;
                        }

                        String formatStr = String.format("COLLECTION:%d,COMMENT:%d,LIKES:%d,VIEWS:%d", col, com, lik, vie);
                        System.out.println("文章的id: " + key);
                        System.out.println("当前时间窗口内的消息处理结果：" + formatStr);
                        return formatStr;
                    }
                }, Materialized.as("hot-atricle-stream-count-001"))
                .toStream()     //转换为kStream
                .map((key, value) -> {
                    return new KeyValue<>(key.key().toString(), formatObj(key.key().toString(), value));    //调用《格式化消息的value数据》方法
                }).to(HotArticleConstants.HOT_ARTICLE_INCR_HANDLE_TOPIC);    //发送消息
        return stream;
    }

    /**
     * 格式化消息的value数据
     */
    public String formatObj(String articleId, String value){
        ArticleVisitStreamMess mess = new ArticleVisitStreamMess();
        mess.setArticleId(Long.valueOf(articleId));   //设置文章id

        //COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0
        String[] valAry = value.split(",");    //对value进行以 "," 分割，结果如：COLLECTION:0
        for (String val : valAry) {
            String[] spl = val.split(":");    //再将结果进行以 ":" 分割   结果如：COLLECTION 和 0
            switch (UpdateArticleMess.UpdateArticleType.valueOf(spl[0])){    //拿到spl中的第[0]参数，结果如：COLLECTION
                case COLLECTION:    //如果第[0]参数是收藏
                    mess.setCollect(Integer.parseInt(spl[1]));    //就把spl中的第[1]个参数赋值给收藏，这个参数就如 COLLECTION:0 中的 0
                    break;
                case COMMENT:    //如果第[0]参数是评论
                    mess.setComment(Integer.parseInt(spl[1]));    //就把spl中的第[1]个参数赋值给评论，这个参数就如 COLLECTION:0 中的 0
                    break;
                case LIKES:    //如果第[0]参数是点赞
                    mess.setLike(Integer.parseInt(spl[1]));    //就把spl中的第[1]个参数赋值给点赞，这个参数就如 COLLECTION:0 中的 0
                    break;
                case VIEWS:    //如果第[0]参数是浏览
                    mess.setView(Integer.parseInt(spl[1]));    //就把spl中的第[1]个参数赋值给浏览，这个参数就如 COLLECTION:0 中的 0
                    break;
            }}
        log.info("聚合消息处理之后的结果为:{}", JSON.toJSONString(mess));
        return JSON.toJSONString(mess);   //把结果转成json字符串返回，因为结果本身是ArticleVisitStreamMess类型，而当前方法的返回值是string类型
    }}